﻿using System.Diagnostics;
using Dapper;
using Oracle.DataAccess.Client;
using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.MessageRepository.Interfaces;
using RockBus.Utilities;
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;

namespace RockBus.MessageRepository.Dapper.Oracle.Db
{
    public class MessageRepository : IMessageRepository
    {
        public MessageRepository(ILogger logger, IDbMessageRepositoryConfiguration configuration)
        {
            this.Logger = logger;
            this.Configuration = configuration;
        }

        public ILogger Logger { get; set; }

        public IDbMessageRepositoryConfiguration Configuration { get; set; }

        public long CreateMessage(
            string correlationId,
            string messageVersion,
            string action,
            DateTime receiveTime,
            string receiveAddress)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                long messageId = 0;
                string dbCmdIns = "insert into Message (CorrelationId " +
                                  "                    ,MessageVersion " +
                                  "                    ,Action " +
                                  "                    ,ReceiveTime " +
                                  "                    ,ReceiveAddress " +
                                  "                    ,CreateTime) " +
                                  "values (@correlationId " +
                                  "       ,@messageVersion " +
                                  "       ,@action " +
                                  "       ,@receiveTime " +
                                  "       ,@receiveAddress " +
                                  "       ,current_timestamp) " +
                                  "returning Id into @Id ";
                messageId = dbConn.QueryReturning<long>(dbCmdIns, new
                {
                    correlationId = correlationId,
                    messageVersion = messageVersion,
                    action = action,
                    receiveTime = receiveTime,
                    receiveAddress = receiveAddress
                }, "Id");

                return messageId;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public long CreateMessageFragment(
            long messageId,
            int partIndex,
            byte[] body,
            int size)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string dbCmdIns = "insert into MessagePart (MessageId " +
                                  "                        ,PartIndex " +
                                 @"                        ,""BODY""
                                                           ,""SIZE"") " +
                                  "values (@messageId " +
                                  "       ,@partIndex " +
                                  "       ,@bodyPrmt " +
                                  "       ,@sizePrmt) " +
                                  "returning Id into @Id";

                byte[] output = new byte[size];
                Buffer.BlockCopy(body, 0, output, 0, size);

                return dbConn.QueryReturning<long>(dbCmdIns, new
                {
                    messageId = messageId,
                    partIndex = partIndex,
                    bodyPrmt = output,
                    sizePrmt = size
                }, "Id");
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public long EnqueueEntry(
            long messageId,
            byte[] routingHeaderByteArray,
            int priority,
            bool orderedDelivery,
            string receiverMachineName,
            int receiverProcessId,
            int receiverAppDomainId,
            long? validityDurationSecs,
            long deliveryAttemptIntervalSecs,
            int maxDeliveryAttemptCount,
            bool isResponse)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string dbCmdIns = "insert into QueueEntry (MessageId " +
                                  "                       ,RoutingHeader " +
                                  "                       ,Priority " +
                                  "                       ,OrderedDelivery " +
                                  "                       ,ReceiverMachineName " +
                                  "                       ,ReceiverProcessId " +
                                  "                       ,ReceiverAppDomainId " +
                                  "                       ,EnqueueTime " +
                                  "                       ,ExpirationTime " +
                                  "                       ,StateId " +
                                  "                       ,DequeueAttemptCount " +
                                  "                       ,DeliveryAttemptCount " +
                                  "                       ,DeliveryAttemptIntervalSecs " +
                                  "                       ,MaxDeliveryAttemptCount " +
                                  "                       ,IsResponse) " +
                                  "values (@messageId " +
                                  "       ,@routingHeader " +
                                  "       ,@priority " +
                                  "       ,@orderedDelivery " +
                                  "       ,@receiverMachineName " +
                                  "       ,@receiverProcessId " +
                                  "       ,@receiverAppDomainId " +
                                  "       ,current_timestamp " +
                                  "       ,current_timestamp + numtodsinterval (@messageValidityDurationSecs, 'second') " +
                                  "       ,0 " + // StateId
                                  "       ,0 " + // DequeueAttemptCount
                                  "       ,0 " + // DeliveryAttemptCount
                                  "       ,@deliveryAttemptIntervalSecs " +
                                  "       ,@maxDeliveryAttemptCount " +
                                  "       ,@isResponse) " +
                                  "returning Id into @Id";

                long queueEntryId = 0;

                long messageVDS = 0;
                if(validityDurationSecs.HasValue)
                    messageVDS = validityDurationSecs.Value;

                queueEntryId = dbConn.QueryReturning<long>(dbCmdIns, new
                {
                    messageId = messageId,
                    routingHeader = routingHeaderByteArray,
                    priority = priority,
                    orderedDelivery = Convertor.BooleanToInt32(orderedDelivery),
                    receiverMachineName = receiverMachineName,
                    receiverProcessId = receiverProcessId,
                    receiverAppDomainId = receiverAppDomainId,
                    messageValidityDurationSecs = messageVDS,
                    deliveryAttemptIntervalSecs = deliveryAttemptIntervalSecs,
                    maxDeliveryAttemptCount = maxDeliveryAttemptCount,
                    isResponse = Convertor.BooleanToInt32(isResponse)
                }, "Id");

                return queueEntryId;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public void CompleteMessage(
            long messageId,
            int partCount,
            int size)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                string dbCmdUpdMsg = "update Message " +
                                     "set PartCount = @partCount " +
                                    @"   ,""SIZE""  = @sizePrmt " +
                                     "where Id = @messageId ";

                dbConn.ExecuteOracleSafe(dbCmdUpdMsg, new
                {
                    partCount = partCount,
                    sizePrmt = size,
                    messageId = messageId
                });

                string dbCmdUpdQue = "update QueueEntry " +
                                     "set StateId = 1 " +
                                     "where MessageId = @messageId " +
                                     "  and StateId   = 0 ";

                dbConn.ExecuteOracleSafe(dbCmdUpdQue, new
                {
                    messageId = messageId
                });

            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();
            }
        }

        public byte[] GetMessageFragment(
            long messageId,
            int partIndex)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string dbCmdSel = "select m.Body " +
                                  "from MessageFragment m with (nolock) " +
                                  "where m.MessageId = @messageId " +
                                  "  and m.PartIndex = @partIndex; ";
                byte [] bArr = null;

                IEnumerable<byte[]> byteList = dbConn.Query<byte[]>(dbCmdSel.Replace("@", ":"), new
                {
                    messageId = messageId,
                    partIndex = partIndex
                });

                foreach(byte [] byteArray in byteList)
                {
                    if(byteArray == null)
                    {
                        this.Logger.LogError(MethodBase.GetCurrentMethod(), "MessageFragment could not be read. MessageId: {0}; PartIndex: {1}.", messageId.ToString(), partIndex.ToString());
                        return null;
                    }

                    bArr = byteArray;
                }

                return bArr;
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public Task<DequeueResult> StartDequeueEnties(
            int maxEntryCount,
            string machineName,
            int processId,
            int appDomainId,
            string processorId)
        {
            try
            {
                return Task.Factory.StartNew<DequeueResult>(() => Dequeue(
                    maxEntryCount,
                    machineName,
                    processId,
                    appDomainId,
                    processorId
                ));
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public DequeueResult Dequeue(
            int maxEntryCount,
            string machineName,
            int processId,
            int appDomainId,
            string processorId)
        {
            Guid batchId = Guid.NewGuid();
            int returnCode = 0;

            List<QueueEntry> queueEntries = new List<QueueEntry>();

            OracleConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            #region query

            string dbCmdUpd = "update QueueEntry " +
                              "set StateId             = 2 " +
                              "   ,MachineName         = @machineName " +
                              "   ,ProcessId           = @processId " +
                              "   ,ProcessorId         = @processorId " +
                              "   ,BatchId             = @batchId " +
                              "   ,DequeueTime         = current_timestamp " +
                              "   ,DequeueAttemptCount = DequeueAttemptCount + 1 " +
                              "where StateId = 1 " +
                              "  and BatchId is null " +
                              "  and DeliveryAttemptCount < MaxDeliveryAttemptCount " +
                              " and (ExpirationTime is null " +
                              "   or ExpirationTime < current_timestamp) " +
                              "and ((IsResponse = 0) " +
                              "  or (IsResponse = 1 " +
                              "  and ReceiverMachineName = @machineName " +
                              "  and ReceiverProcessId = @processId " +
                              "  and ReceiverAppDomainId = @appDomainId)) ";

            string dbCmdSel = "select q.Id " +
                              "      ,q.MessageId " +
                              "      ,q.ExpirationTime " +
                              "      ,q.StateId " +
                              "      ,q.RoutingHeader " +
                              "      ,q.Priority " +
                              "      ,q.OrderedDelivery " +
                              "      ,q.EnqueueTime " +
                              "      ,q.ReceiverMachineName " +
                              "      ,q.ReceiverProcessId " +
                              "      ,q.ReceiverAppDomainId " +
                              "      ,q.MachineName " +
                              "      ,q.ProcessId " +
                              "      ,q.ProcessorId " +
                              "      ,q.BatchId " +
                              "      ,q.DequeueTime " +
                              "      ,q.DequeueAttemptCount " +
                              "      ,q.DeliveryAttemptCount " +
                              "      ,q.IsResponse " +
                              "      ,m.PartCount " +
                             @"      ,m.""SIZE"" " +
                              "      ,m.CorrelationId " +
                              "      ,m.MessageVersion" +
                              "      ,m.Action " +
                              "      ,m.ReceiveTime " +
                              "      ,m.ReceiveAddress " +
                              "      ,m.CreateTime " +
                              "from QueueEntry q " +
                              "    ,Message m " +
                              "where q.BatchId = @batchId " +
                              "  and q.MessageId = m.Id " +
                              "order by q.Priority desc " +
                              "        ,q.Id asc ";
            #endregion

            dbConn.Open();
            OracleTransaction dbTrans = dbConn.BeginTransaction(System.Data.IsolationLevel.ReadCommitted);
            try
            {
                dbConn.ExecuteOracleSafe(dbCmdUpd, new
                {
                    count = maxEntryCount,
                    machineName = machineName,
                    processId = processId,
                    appDomainId = appDomainId,
                    processorId = processorId,
                    batchId = batchId.ToString()
                }, dbTrans);

                dbTrans.Commit();
                // end of transaction

                IEnumerable<dynamic> qeList = dbConn.Query(dbCmdSel.Replace("@", ":"), new
                {
                    batchId = batchId.ToString()
                });

                if(qeList.Count() == 0)
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "No entries to dequeue.");
                    return null;
                }

                foreach(dynamic qEntry in qeList)
                {
                    QueueEntry qe = new QueueEntry();

                    qe.Id = qEntry.ID;
                    qe.MessageId = qEntry.MESSAGEID;
                    qe.ExpirationTime = qEntry.EXPIRATIONTIME;
                    qe.StateId = (byte) qEntry.STATEID;
                    qe.RoutingHeader = RoutingHeader.CreateFromByteArray((byte[])qEntry.ROUTINGHEADER);
                    qe.Priority = (int) qEntry.PRIORITY;
                    qe.OrderedDelivery = Convertor.Int32ToBoolean (qEntry.ORDEREDDELIVERY);
                    qe.EnqueueTime = qEntry.ENQUEUETIME;
                    qe.ReceiverMachineName = qEntry.RECEIVERMACHINENAME;
                    qe.ReceiverProcessId = (int) qEntry.RECEIVERPROCESSID;
                    qe.ReceiverAppDomainId = (int) qEntry.RECEIVERAPPDOMAINID;
                    qe.MachineName = qEntry.MACHINENAME;
                    qe.ProcessId = (int) qEntry.PROCESSID;
                    qe.ProcessorId = qEntry.PROCESSORID;
                    qe.BatchId = Guid.Parse((string) qEntry.BATCHID);
                    qe.DequeueTime = qEntry.DEQUEUETIME;
                    qe.DequeueAttemptCount = (int) qEntry.DEQUEUEATTEMPTCOUNT;
                    qe.DeliveryAttemptCount = (int) qEntry.DELIVERYATTEMPTCOUNT;
                    qe.IsResponse = Convertor.Int32ToBoolean (qEntry.ISRESPONSE);
                    qe.PartCount = (int) qEntry.PARTCOUNT;
                    qe.Size = (int) qEntry.SIZE;
                    qe.CorrelationId = qEntry.CORRELATIONID;
                    qe.MessageVersion = qEntry.MESSAGEVERSION;
                    qe.Action = qEntry.ACTION;
                    qe.ReceiveTime = qEntry.RECEIVETIME;
                    qe.ReceiveAddress = qEntry.RECEIVEADDRESS;
                    qe.CreateTime = qEntry.CREATETIME;

                    queueEntries.Add(qe);
                }
            }
            catch(Exception ex)
            {
                dbTrans.Rollback();
                batchId = Guid.Empty;
                returnCode = 1;
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "Exception-msg: {0}", ex.Message);
            }
            finally
            {
                if(dbConn != null)
                    dbConn.Dispose();

                if(dbTrans != null)
                    dbTrans.Dispose();
            }

            if(returnCode != 0)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "Return code was {0}.", returnCode);
                return null;
            }

            if(batchId == null || batchId == Guid.Empty)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), "BatchId was NULL.");
                return null;
            }

            DequeueResult result = new DequeueResult(batchId, queueEntries);
            return result;
        }

        public void CompleteEntry(
            long queueEntryId)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                string dbCmdUpd = "update q with (updlock) " +
                                  "set q.StateId = 3 " +
                                  "   ,q.DeliveryAttemptCount = DeliveryAttemptCount + 1 " +
                                  "   ,q.DeliveryTime = current_timestamp " +
                                  "from QueueEntry q " +
                                  "where q.Id = @queueEntryId ";
                dbConn.ExecuteOracleSafe(dbCmdUpd, new
                {
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void RollbackEntry(
            long queueEntryId,
            Exception exception)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();

                string dbCmdUpd = "update q with (rowlock) " +
                                  "set q.StateId = 1 " +
                                  "   ,q.MachineName = null " +
                                  "   ,q.ProcessId = null " +
                                  "   ,q.ProcessorId = null " +
                                  "   ,q.BatchId = null " +
                                  "   ,q.DeliveryAttemptCount = q.DeliveryAttemptCount + 1 " +
                                  "   ,q.DeliveryTime = null " +
                                  "   ,q.ErrorMessage = @errorMsg " +
                                  "from QueueEntry q " +
                                  "where q.Id = @queueEntryId ";

                dbConn.ExecuteOracleSafe(dbCmdUpd, new
                {
                    errorMsg = exception.ToString(),
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void FaultEntry(
            long queueEntryId,
            Exception exception)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);

            try
            {
                dbConn.Open();
                string dbCmdUpd = "update q with (rowlock) " +
                                  "set q.StateId = 4 " +
                                  "   ,q.BatchId = null " +
                                  "   ,q.DeliveryAttemptCount = q.DeliveryAttemptCount + 1 " +
                                  "   ,q.DeliveryTime = null " +
                                  "   ,q.ErrorMessage = @errorMsg " +
                                  "from QueueEntry q " +
                                  "where q.Id = @queueEntryId ";

                dbConn.ExecuteOracleSafe(dbCmdUpd, new
                {
                    errorMsg = exception.ToString(),
                    queueEntryId = queueEntryId
                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void CreateErrorMessage(
            long messageId,
            byte[] routingHeaderByteArray,
            string machineName,
            int processId,
            int appDomainId,
            int deliveryAttemptCount,
            Exception exception)
        {
            IDbConnection dbConn = new OracleConnection(this.Configuration.ConnectionString);
            try
            {
                dbConn.Open();

                string dbCmdIns = "insert into ErrorMessage (MessageId " +
                                  "                         ,RoutingHeader " +
                                  "                         ,MachineName " +
                                  "                         ,ProcessId " +
                                  "                         ,AppDomainId " +
                                  "                         ,DeliveryAttemptCount " +
                                  "                         ,ErrorMessage " +
                                  "                         ,CreateTime) " +
                                  "values (@messageId " +
                                  "       ,@routingHeader " +
                                  "       ,@machineName " +
                                  "       ,@processId " +
                                  "       ,@appDomainId " +
                                  "       ,@deliveryAttemptCount " +
                                  "       ,@errorMsg " +
                                  "       ,current_timestamp) ";

                dbConn.ExecuteOracleSafe(dbCmdIns, new
                {
                    messageId = messageId,
                    routingHeader = routingHeaderByteArray,
                    machineName = machineName,
                    processId = processId,
                    appDomainId = appDomainId,
                    deliveryAttemptCount = deliveryAttemptCount,
                    errorMsg = exception.ToString()

                });
            }
            catch(Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        private string ToDelimitedString<T>(IEnumerable<T> items, char seperator = ',')
        {
            StringBuilder delimitedString = new StringBuilder();
            foreach(var item in items)
            {
                delimitedString.Append(item.ToString());
                delimitedString.Append(seperator);
            }
            return delimitedString.ToString();
        }

        private IEnumerable<string> FromDelimitedString(string delimitedString, char seperator = ',')
        {
            string[] fragments = delimitedString.Split(seperator);
            foreach(var fragment in fragments)
                yield return fragment;
        }
    }
}